package cn.demoncat.util.mq.util;

import cn.demoncat.util.lang.JsonUtil;
import cn.demoncat.util.lang.StringUtil;
import cn.demoncat.util.lang.constant.StringConstant;
import cn.demoncat.util.mq.constant.MqLogMsg;
import cn.demoncat.util.mq.entity.*;
import cn.demoncat.util.mybatis.util.TransactionUtil;
import cn.demoncat.util.web.config.AppProperties;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

import javax.annotation.Resource;

/**
 * Mq工具类
 *
 * 1、子类（Kafka、RabbitMQ）实现send()和sendDelay()并声明为Bean("mqUtil")
 * 2、消息发出后，无法删除，需要在消费者端做处理
 * 
 * @author 延晓磊
 *
 * @since 2018年8月10日
 */
public abstract class MqUtil {

	@Resource
	private AppProperties appProperties;

	/**
	 * 主题前缀
	 *
	 * 配置：spring.application.isolator
	 * 场景：多个系统，使用同一个MQ，为了防止主题冲突，分别对主题添加前缀来标识
	 * 发送：判断spring.application.isolator非空时，添加前缀 > mqUtil.getTopic（如果topic前缀*则忽略）
	 * 接收：判断spring.application.isolator非空时，添加前缀 > MqUtil.getTopicExpression
	 */
	private String prefix = null;

	/**
	 * 发送消息（子类实现，内部调用）
	 *
	 * @param topic		主题
	 * @param key		Key：均衡partition，为空时获取msg.id
	 * @param msg		消息（MqMessage）
	 * @param callback	回调
	 *
	 * @author 延晓磊
	 * @since 2021年02月23日
	 */
	protected abstract void send(String topic, String key, MqMessage msg, MqCallback callback);

	/**
	 * 发送延时消息（子类实现，内部调用）
	 *
	 * @param message
	 * 
	 * @author 延晓磊
	 * @since 2021年02月23日
	 */
	protected abstract void sendDelay(MqDelayMessage message);

	/**
	 * 监控延时消息（子类实现，配置调用）
	 *
	 * 1、对于不支持延迟消息的MQ（如KAFKA），需要借助外部数据库暂存，因此需要监控暂存的消息并补发
	 * 2、使用默认的全局延时队列
	 *
	 * @author 延晓磊
	 * @since 2021年02月23日
	 */
	public void monitorDelayMessage(){ }

	/**
	 * 监控延时消息（子类实现，配置调用）
	 *
	 * 1、对于不支持延迟消息的MQ（如KAFKA），需要借助外部数据库暂存，因此需要监控暂存的消息并补发
	 * 2、使用自定义的全局延时队列
	 *
	 * @param name	队列名称
	 * @param seconds 消费间隔
	 *
	 * @author 延晓磊
	 * @since 2021年02月23日
	 */
	public void monitorDelayMessage(String name, int seconds){ }

	/**
	 * 获取消息对象
	 * 
	 * @param msg
	 * @param clazz
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年8月9日
	 */
	public static <T>T get(MqMessage msg, Class<T> clazz) {
		return JsonUtil.parse(msg.getMsg(), clazz);
	}
	
	/**
	 * 接收消息，转换为Message
	 * 
	 * @param content
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年8月9日
	 */
	public static MqMessage get(String content) {
		return JsonUtil.parse(content, MqMessage.class);
	}
	
	/**
	 * 发送消息
	 *
	 * 默认callback记录发送日志
	 * 
	 * @param topic		主题
	 * @param key		Key：均衡partition，为空时获取msg.id
	 * @param msg		消息
	 * @param logger	日志
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void send(String topic,Long key,MqMessage msg,Logger logger) {
		 send(topic, StringUtil.toString(key), msg, logger);
	}

	/**
	 * 发送消息
	 * 
	 * @param topic	主题
	 * @param key	Key：均衡partition，为空时获取msg.id
	 * @param msg	消息
	 * @param callback	成功/失败回调
	 * @param logger	日志
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void send(String topic,Long key,MqMessage msg,MqCallback callback,Logger logger) {
		send(topic, StringUtil.toString(key), msg, callback, logger);
	}

	/**
	 * 发送消息
	 *
	 * 默认callback记录发送日志
	 *
	 * @param topic		主题
	 * @param key		Key：均衡partition，为空时获取msg.id
	 * @param msg		消息
	 * @param logger	日志
	 * @return
	 *
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void send(String topic,String key,MqMessage msg,Logger logger) {
		send(topic, key, msg,null, logger);
	}

	/**
	 * 发送消息
	 *
	 * @param topic		主题
	 * @param key		Key：均衡partition，为空时获取msg.id
	 * @param msg		消息
	 * @param callback	成功/失败回调，默认为DefaultMqCallback
	 * @param logger	日志
	 * @return
	 *
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void send(String topic,String key,MqMessage msg,MqCallback callback,Logger logger) {
		// 补全主题
		topic = this.getTopic(topic);
		// 发送日志
		logger.info(JsonUtil.toStr(new MqLog(MqLogMsg.SEND, topic, msg)));
		// KEY
		key = msg.key(key);
		// 回调
		if (callback == null) {
			callback = new DefaultMqCallback();
		}
		callback.setContent(msg);
		callback.setKey(key);
		callback.setTopic(topic);
		callback.setLog(logger);
		// 发送消息（子类实例）
		send(topic, key, msg, callback);
	}

	/**
	 * 发送延时消息
	 *
	 * @param msg		消息
	 * @param logger	日志
	 * @return
	 *
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void sendDelay(MqDelayMessage msg, Logger logger) {
		// 补全主题
		msg.setTopic(this.getTopic(msg.getTopic()));
		// 发送日志
		logger.info(JsonUtil.toStr(new MqLog(MqLogMsg.SEND_DELAY, msg.getTopic(), msg.getMessage())));
		// 发送消息（子类实例）
		sendDelay(msg);
	}

	/**
	 * 事务提交后发送消息
	 * 
	 * 默认callback记录发送日志
	 * 
	 * @param topic	主题
	 * @param key	Key：均衡partition，为空时获取msg.id
	 * @param msg	消息
	 * @param logger	日志
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void txSend(String topic,Long key,MqMessage msg,Logger logger) {
		 TransactionUtil.afterCommit(() -> send(topic, key, msg, logger));
	}

	/**
	 * 事务提交后发送消息
	 *
	 * @param topic	主题
	 * @param key	Key：均衡partition，为空时获取msg.id
	 * @param msg	消息
	 * @param callback	成功/失败回调
	 * @param logger	日志
	 * @return
	 *
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void txSend(String topic,Long key,MqMessage msg,MqCallback callback,Logger logger) {
		TransactionUtil.afterCommit(() -> send(topic, key, msg, callback, logger));
	}

	/**
	 * 事务提交后发送消息
	 *
	 * 默认callback记录发送日志
	 *
	 * @param topic	主题
	 * @param key	Key：均衡partition，为空时获取msg.id
	 * @param msg	消息
	 * @param logger	日志
	 * @return
	 *
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void txSend(String topic,String key,MqMessage msg,Logger logger) {
		TransactionUtil.afterCommit(() -> send(topic, key, msg,null, logger));
	}

	/**
	 * 事务提交后发送消息
	 * 
	 * @param topic	主题
	 * @param key	Key：均衡partition，为空时获取msg.id
	 * @param msg	消息
	 * @param callback	成功/失败回调
	 * @param logger	日志
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void txSend(String topic,String key,MqMessage msg,MqCallback callback,Logger logger) {
		TransactionUtil.afterCommit(() -> send(topic, key, msg, callback, logger));
	}

	/**
	 * 事务提交后发送延时消息
	 *
	 * 默认callback记录发送日志
	 *
	 * @param msg	消息
	 * @param logger	日志
	 * @return
	 *
	 * @author 延晓磊
	 *
	 * @since 2018年8月10日
	 */
	public void txSendDelay(MqDelayMessage msg,Logger logger) {
		TransactionUtil.afterCommit(() -> sendDelay(msg, logger));
	}

	/**
	 * 获取主题
	 *
	 * 1、通过添加前缀，实现多系统隔离（如果topic前缀*则忽略）
	 * 2、实例发送消息前调用本方法，来补全主题名
	 * 3、接收方须拼接相同的前缀
	 *
	 * @param topic
	 * @return
	 *
	 * @author 延晓磊
	 * @since 2023年04月23日
	 */
	public String getTopic(String topic){
		if (topic.startsWith(StringConstant.STAR)) {
			// 忽略隔离前缀
			return topic.substring(1);
		}
		return this.getPrefix().isEmpty() ? topic : (topic.startsWith(this.getPrefix()) ? topic : (this.getPrefix() + topic));
	}

	/**
	 * 获取主题表达式（拼接前缀）
	 *
	 * @param topic 声明主题
	 * @return 主题表达式，例 "#{mqUtil.getTopic('${spring.application.name}-job')}" > prefix-app-job
	 *
	 * @author 延晓磊
	 * @since 2023年04月27日
	 */
	public static String getTopicExpression(String topic){
		return "#{mqUtil.getTopic('" + topic + "')}";
	}

	/**
	 * 获取主题前缀
	 *
	 * @return
	 *
	 * @author 延晓磊
	 * @since 2023年04月23日
	 */
	private String getPrefix(){
		if (this.prefix == null) {
			if (StringUtils.isBlank(appProperties.getIsolator())) {
				// ""
				this.prefix = StringConstant.EMPTY;
			}else{
				// isolator-
				this.prefix = appProperties.getIsolator().toLowerCase() + StringConstant.BAR;
			}
		}
		return this.prefix;
	}
}
